package com.fanxuankai.kafka.connect.transforms.predicates;

import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.connect.connector.ConnectRecord;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.transforms.predicates.Predicate;
import org.apache.kafka.connect.transforms.util.SimpleConfig;

import java.util.Map;
import java.util.Objects;

import static org.apache.kafka.connect.transforms.util.Requirements.requireMapOrNull;
import static org.apache.kafka.connect.transforms.util.Requirements.requireStructOrNull;

/**
 * 逻辑删除谓语
 *
 * @author fanxuankai
 */
public abstract class LogicDeleted<R extends ConnectRecord<R>> implements Predicate<R> {
    public static final String OVERVIEW_DOC = "Record logic deleted.";
    private static final String REGEX_CONFIG = "regex";
    private static final String VALUE_CONFIG = "value";
    public static final ConfigDef CONFIG_DEF = new ConfigDef()
            .define(VALUE_CONFIG, ConfigDef.Type.STRING, ConfigDef.NO_DEFAULT_VALUE,
                    new ConfigDef.NonEmptyString(), ConfigDef.Importance.MEDIUM,
                    "The field value.")
            .define(REGEX_CONFIG, ConfigDef.Type.STRING, ConfigDef.NO_DEFAULT_VALUE,
                    new ConfigDef.NonEmptyString(), ConfigDef.Importance.MEDIUM,
                    "The field regex.");

    private static final String PURPOSE = "logic deleted";
    private String fieldRegex;
    private String fieldValue;

    @Override
    public ConfigDef config() {
        return CONFIG_DEF;
    }

    @Override
    public boolean test(R record) {
        if (operatingValue(record) == null) {
            return true;
        }
        Schema schema = operatingSchema(record);
        String logicValue = null;
        if (schema == null) {
            Map<String, Object> value = requireMapOrNull(operatingValue(record), PURPOSE);
            for (Map.Entry<String, Object> entry : value.entrySet()) {
                if (entry.getKey().toLowerCase().matches(fieldRegex)) {
                    logicValue = entry.getValue().toString();
                }
            }
        } else {
            Struct value = requireStructOrNull(operatingValue(record), PURPOSE);
            for (Field field : schema.fields()) {
                if (field.name().toLowerCase().matches(fieldRegex)) {
                    logicValue = value.get(field).toString();
                }
            }
        }
        if (logicValue == null) {
            return false;
        }
        return Objects.equals(logicValue, fieldValue);
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {
        SimpleConfig simpleConfig = new SimpleConfig(config(), configs);
        fieldRegex = simpleConfig.getString(REGEX_CONFIG).toLowerCase();
        fieldValue = simpleConfig.getString(VALUE_CONFIG);
    }

    protected abstract Schema operatingSchema(R record);

    protected abstract Object operatingValue(R record);

    protected abstract R newRecord(R record, Schema updatedSchema, Object updatedValue);

    public static class Key<R extends ConnectRecord<R>> extends RecordPredicate<R> {
        @Override
        protected Schema operatingSchema(R record) {
            return record.keySchema();
        }

        @Override
        protected Object operatingValue(R record) {
            return record.key();
        }

        @Override
        protected R newRecord(R record, Schema updatedSchema, Object updatedValue) {
            return record.newRecord(record.topic(), record.kafkaPartition(), updatedSchema, updatedValue,
                    record.valueSchema(), record.value(), record.timestamp());
        }
    }

    public static class Value<R extends ConnectRecord<R>> extends LogicDeleted<R> {
        @Override
        protected Schema operatingSchema(R record) {
            return record.valueSchema();
        }

        @Override
        protected Object operatingValue(R record) {
            return record.value();
        }

        @Override
        protected R newRecord(R record, Schema updatedSchema, Object updatedValue) {
            return record.newRecord(record.topic(), record.kafkaPartition(), record.keySchema(), record.key(),
                    updatedSchema, updatedValue, record.timestamp());
        }
    }
}
